package org.huangrui.spark.scala.streaming

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * @author hr
 * @create 2020-12-27 1:27 
 */
object SparkStreaming13_Req3 {
  def main(args: Array[String]): Unit = {
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")
    val sc: StreamingContext = new StreamingContext(sparkConf, Seconds(5))

    //    1.定义 Kafka 参数
    val kafkaPara: Map[String, Object] = Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "hadoop210:9092,hadoop211:9092,hadoop212:9092",
      ConsumerConfig.GROUP_ID_CONFIG -> "spark",
      "key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer",
      "value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer"
    )
    //    2.读取 Kafka 数据创建 DStream
    val kafkaDStream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](sc,
      LocationStrategies.PreferConsistent,//采集的节点和计算的节点该如何做匹配，类似于spark core中的首选位置
      ConsumerStrategies.Subscribe[String, String](Set("spark"), kafkaPara))
    val adclickData: DStream[AdClickData] = kafkaDStream.map {
      kafkaData => {
        val data: String = kafkaData.value()
        val datas: Array[String] = data.split(" ")
        AdClickData(datas(0), datas(1), datas(2), datas(3), datas(4))
      }
    }
    // 最近一分钟，每10秒计算一次
    // 12:01 => 12:00
    // 12:11 => 12:10
    // 12:19 => 12:10
    // 12:25 => 12:20
    // 12:59 => 12:50

    // 55 => 50, 49 => 40, 32 => 30
    // 55 / 10 * 10 => 50
    // 49 / 10 * 10 => 40
    // 32 / 10 * 10 => 30
    // 这里涉及窗口的计算
    val reduceDs: DStream[(Long, Int)] = adclickData.map {
      data => {
        val ts: Long = data.ts.toLong
        val newTs: Long = ts / 10000 * 10000
        (newTs, 1)
      }
    }.reduceByKeyAndWindow((x:Int,y:Int)=>{x+y}, Seconds(60), Seconds(10))
    reduceDs.print()



    sc.start()
    sc.awaitTermination()
  }
  // 广告点击数据
  case class AdClickData( ts:String, area:String, city:String, user:String, ad:String )
}
